/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.ratings.normilize;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.processing.base.mapreduce.standard.mappers.ArtistToUserValueMatrixMapper;
import aiaudio.processing.base.mapreduce.standard.reducers.MatrixTransponationReducer;
import aiaudio.lastfm.hbase.CannotCreateTableException;
import aiaudio.Application;
import aiaudio.processing.MainTableGroup;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.mapreduce.Job;

/**
 *
 * @author nastya
 */
public class RatingRecalculatorBase {
    protected MainTableGroup tablePackage;

    protected String inputTableName;
    protected Class<? extends TableMapper<ImmutableBytesWritable, CoordinateDoubleValueKey>> mapperClass;
    protected Class<? extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable>> reducerClass;
    private String temporaryTableName;

    public RatingRecalculatorBase(MainTableGroup tablePackage, RatingAlgorithmDefinition algorithmDefinition) {
        this.tablePackage = tablePackage;
        inputTableName = algorithmDefinition.getInputTableName();
        mapperClass = algorithmDefinition.getMapperClass();
        reducerClass = algorithmDefinition.getReducerClass();
    }

    public void start() throws IOException, InterruptedException, ClassNotFoundException, CannotCreateTableException{
        Job job = createProcessingJob();
        job.waitForCompletion(true);
        job = createConversionJob();
        job.waitForCompletion(true);
    }
    
    protected Job createProcessingJob() throws IOException, CannotCreateTableException {
        Job job = new Job();

        Scan scan = new Scan();
        
        String ratingsTableName = Application.database().getByName(inputTableName).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(ratingsTableName, scan, mapperClass, ImmutableBytesWritable.class, CoordinateDoubleValueKey.class, job);

        String temporaryTable = Application.database().getByName(temporaryTableName).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(temporaryTable, reducerClass, job);

        return job;
    }
    
    protected Job createConversionJob() throws IOException, CannotCreateTableException {
        Job job = new Job();

        Scan scan = new Scan();
        
        String temporaryTable = Application.database().getByName(temporaryTableName).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(temporaryTable, scan, ArtistToUserValueMatrixMapper.class, ImmutableBytesWritable.class, CoordinateDoubleValueKey.class, job);
        
        String ratingsTableName = Application.database().getByName(tablePackage.getPrepareRatingsAlgTableGroup().getNormilizedRatingMatrix()).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(ratingsTableName, MatrixTransponationReducer.class, job);
        
        return job;
    }


   
    public void prepeare() throws IOException, CannotCreateTableException, InterruptedException, ClassNotFoundException {
        temporaryTableName = Application.database().createTemporaryMatrixTable();
    }
    
}
